package com.dingtone.kafkaclient;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Properties;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import com.alibaba.fastjson.JSON;
import com.dingtone.Msg.Msg;
import com.dingtone.serialization.JsonDeserializer;
import com.dingtone.serialization.JsonSerializer;

class Config {
	public String Topic;
	public String Group;
	public int Partition;
	public String ZkClient;
	public String BootStrapServer;
}

public class MsgConsumer {
	private final int MQ_SIZE = 1000;
	private final int MQ_POLL_SIZE = 1000;
	private final KafkaConsumer consumer;
	private Config config;
	private ExecutorService executor;
	private volatile Boolean closed = false;

	private Properties getProps(Config cfg) {
		Properties props = new Properties();
		//JsonSerializer<Msg> json = new JsonSerializer<Msg>();
		//JsonDeserializer jsonDeserializer = new JsonDeserializer();
		props.put("bootstrap.servers", "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094");
		props.put("group.id", cfg.Group);
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		//props.put("value.deserializer", jsonDeserializer.getClass());
		return props;
	}

	public MsgConsumer(Config cfg) {
		this.config = cfg;
		consumer = new KafkaConsumer(getProps(cfg),
				new StringDeserializer(),new JsonDeserializer(Msg.class));
		int cpuNums = Runtime.getRuntime().availableProcessors();
		executor = Executors.newFixedThreadPool(cpuNums * 2);
	}

	public void shutdown() {
		synchronized (closed) {
			closed = true;
		}
	}

	// @SuppressWarnings("unchecked")
	public void run(int numThreads) {
		Future f = executor.submit(new Runnable() {
			public void run() {
				List<TopicPartition> tps = new ArrayList<TopicPartition>();
				tps.add(new TopicPartition(config.Topic, 0));
				tps.add(new TopicPartition(config.Topic, 1));
				tps.add(new TopicPartition(config.Topic, 2));
				System.out.println("***[MQConsumer]*** consumer " + config.Topic + "\n");
				//consumer.assign(tps);
				consumer.subscribe(Arrays.asList(config.Topic));
				// TODO Auto-generated method stub
				while (true) {
					synchronized (closed) {
						if (closed) {
							System.out.println("!!![System quit]!!!!");
							break;
						}
					}
					ConsumerRecords<String, Msg> records = consumer.poll(MQ_POLL_SIZE);
					System.out.println("!!![Poll loop]!!!!");
					List<Msg> tRecs = new ArrayList<Msg>();
					// Collections.sort();
					for (ConsumerRecord<String, Msg> record : records) {
						//record.topic()
						System.out.println("!!!![Record]!!! partition:"+ record.partition()
												+" offset:"+record.offset()+"\n");
						Msg msg = record.value();
						tRecs.add(msg);
						// System.out.println("***[MQClient]**** receive
						// from:"+record.topic() +"
						// partition:"+record.partition()+"
						// value:"+record.value()+"\n");
					}
					//consumer.commitSync();

					Collections.sort(tRecs, new Comparator<Msg>() {
						public int compare(Msg o1, Msg o2) {
							if (o1.Ticket == o2.Ticket) {
								return (int) (o1.Time - o2.Time);
							}
							return (int) (o1.Ticket - o2.Ticket);
						}
					});

					for (Msg msg : tRecs) {
						System.out.println("***[MQClient]**** receive MSG:" + msg.Content + " ticket:" + msg.Ticket
								+ " Time:" + msg.Time + "\n");
					}
				}
				consumer.unsubscribe();
			}
		});
	}

	@SuppressWarnings("deprecation")
	public static void main(String[] arg) {
		  CommandLineParser parser = new PosixParser( );  
		  Options options = new Options( );  
		  options.addOption("h", "help", false, "Print this usage information");  
		  options.addOption("p", "Partition", true, "Topic partition" );  
		  options.addOption("t", "Topic", true, "Kafka consumer topic");
		  options.addOption("g", "Group", true, "Kafka consumer group name");
		  options.addOption("zk", "Zookeeper", true, "Zookeeper settings");

		  Config cfg = new Config();
		  
		  HelpFormatter hf = new HelpFormatter();
		  hf.printHelp("testApp", options, true);
		  hf.setWidth(110);
		  try {
			CommandLine cmd = parser.parse(options, arg);
			if (cmd.hasOption('h')) {
                //打印使用帮助
                hf.printHelp("testApp", options, true);
            }
			else{
				//启动服务地址
				if(cmd.hasOption("bs")){
					cfg.BootStrapServer = cmd.getOptionValue("bs");
				}
				else{
					cfg.BootStrapServer = "127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094";
				}
				//分区id
				if(cmd.hasOption('p')){
					cfg.Partition = Integer.parseInt(cmd.getOptionValue('p'));
				}else{
					cfg.Partition = 0;
				}
				//topic
				if(cmd.hasOption('t')){
					cfg.Topic = cmd.getOptionValue('t');
				}
				else{
					cfg.Topic ="page_visits0";
				}
				//group
				if(cmd.hasOption('g')){
					cfg.Group = cmd.getOptionValue('g');
				}
				else{
					cfg.Group = "group-1";
				}
				//zkclient
				if(cmd.hasOption("zk")){
					cfg.ZkClient = cmd.getOptionValue("zk");
				}else{
					cfg.ZkClient = "127.0.0.1:2181";
				}
			}
		} catch (ParseException e) {
			// TODO Auto-generated catch block
			hf.printHelp("testApp", options, true);
			return;
		}
		  
		MsgConsumer demo = new MsgConsumer(cfg);
		demo.run(12);

		while(true){
			String strInput = readString();
			if(strInput.compareTo("quit") ==0){
				break;
			}
		} 
		demo.shutdown();
	}

	public static String readString() {
		try {
			return new BufferedReader(new InputStreamReader(System.in)).readLine();
		} catch (Exception e) {
			e.printStackTrace();
			return "";
		} finally {

		}

	}

}
